package com.codeboy.mediasoupandroiddemo.socket;

import android.util.Log;

import com.codeboy.mediasoupandroiddemo.utils.HttpsConfig;
import com.codeboy.mediasoupandroiddemo.utils.RandomString;

import org.java_websocket.handshake.ServerHandshake;
import org.json.JSONException;
import org.json.JSONObject;

import java.net.URI;
import java.net.URISyntaxException;
import java.util.concurrent.Callable;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import javax.net.SocketFactory;

import okhttp3.OkHttpClient;

/**
 * @Author: Codeboy 2685312867@qq.com
 * @Date: 2020-04-04 10:17
 */

public class EchoSocket implements MessageObserver.Subscriber, WSocketClientDelegate {
    private static final String TAG = "EchoSocket";

    private final CopyOnWriteArraySet<MessageObserver.Observer> mObservers;
    private CountDownLatch mLatch;
    private final ExecutorService mExecutorService;

    private WSocketClient mSocket;

    public EchoSocket() {
        mObservers = new CopyOnWriteArraySet<>();
        mExecutorService = Executors.newSingleThreadExecutor();
    }

    public Future<Void> connect(String wsUrl) {
        Log.d(TAG, "connect wsUrl=" + wsUrl);
        if (!wsUrl.startsWith("ws://") && !wsUrl.startsWith("wss://")) {
            throw new RuntimeException("Socket url must start with ws/wss");
        }

        if (mSocket != null) {
            throw new IllegalStateException("Socket is already defined");
        }

        try {

            mSocket = new WSocketClient(new URI(wsUrl), WSocketClient.getDraftInfo());
            HttpsConfig.trustAllHosts(mSocket);

        } catch (URISyntaxException e) {
            e.printStackTrace();
        }

        this.mSocket.delegate = this;
        this.mSocket.connect();

        mLatch = new CountDownLatch(1);

        Callable<Void> callable = () -> {
            MessageObserver.Observer observer = (method, id, notification, data) -> {
                Log.d(TAG, "GOT EVENT " + method);
                if (method.equals("open")) {
                    mLatch.countDown();
                }
            };

            mObservers.add(observer);

            mLatch.await();
            Log.d(TAG, "Connected remove obs");
            mObservers.remove(observer);
            return null;
        };

        return mExecutorService.submit(callable);
    }


    public void sendMessage(String method, JSONObject body) {
        sendMessage(method, body, RandomString.numlength(7));
    }

    public void sendMessage(String method, JSONObject body, int requestId) {
        JSONObject json = new JSONObject();
        try {
            json.put("request", true);
            json.put("id", requestId);
            json.put("method", method);
            json.put("data", body);
            send(json);
        } catch (JSONException e) {
            e.printStackTrace();
        }

    }

    public void send(JSONObject message) {
        Log.i("send", message.toString());
        mSocket.send(message.toString());
    }

    public Future<JSONObject> sendWithFuture(String method, JSONObject body) {
        return new AckCall(method).sendAckRequest(body);
    }


    @Override
    public void onOpen(ServerHandshake handshakedata) {
        notifyObservers(ActionEvent.OPEN, 0, false, null);
    }

    @Override
    public void onMessage(String message) {
        Log.d(TAG, "onMessage text=" + message);
        try {
            JSONObject jsonObject = new JSONObject(message);
            int id = -1;
            if (!jsonObject.isNull("id")) {
                id = jsonObject.getInt("id");
            }
            String method = getJsonString(jsonObject, "method");
            boolean notification = false;
            if (!jsonObject.isNull("notification")) {
                notification = jsonObject.getBoolean("notification");
            }
            JSONObject data = jsonObject.getJSONObject("data");
            notifyObservers(method, id, notification, data);
        } catch (JSONException e) {
            Log.e(TAG, "Failed to handle message", e);
            e.printStackTrace();
        }


    }

    private String getJsonString(JSONObject jsonObject, String name) throws JSONException {
        String s = "";
        if (!jsonObject.isNull(name)) {
            s = jsonObject.getString(name);
        }
        return s;
    }

    @Override
    public void onClose(int code, String reason, boolean remote) {
        mSocket = null;
        Log.e("TAG","code = " +  code);
    }

    @Override
    public void onError(Exception ex) {

        Log.e("TAG","ex = " +ex.toString());

    }

    @Override
    public void register(MessageObserver.Observer observer) {
        mObservers.add(observer);
    }

    @Override
    public void unregister(MessageObserver.Observer observer) {
        mObservers.remove(observer);
    }

    @Override
    public void notifyObservers(String method, int id, boolean notification, JSONObject data) {
        for (final MessageObserver.Observer observer : mObservers) {
            observer.on(method, id, notification, data);
        }
    }


    private class AckCall {
        private final String method;
        private final int id;

        private JSONObject mResponse;

        AckCall(String method) {
            id = RandomString.numlength(7);
            this.method = method;
        }

        Future<JSONObject> sendAckRequest(JSONObject body) {
            mLatch = new CountDownLatch(1);

            Callable<JSONObject> callable = () -> {
                MessageObserver.Observer observer = (method, id, notification, data) -> {
                    if (this.id == id) {
                        // Acknowledgement received
                        mResponse = data;
                        mLatch.countDown();
                    } else {
                        Log.d(TAG, "Another event 1" + id);
                    }
                };

                mObservers.add(observer);
                sendMessage(method, body, id);
                mLatch.await();
                mObservers.remove(observer);
                return mResponse;
            };
            return mExecutorService.submit(callable);
        }
    }
}
